package com.blink.zookeeper.operate;

import com.blink.base.TestZK;
import com.blink.zookeeper.ZooKeeperCreator;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * <Description> 读取数据</Description>
 * <ClassName> TestZKQueryData</ClassName>
 *
 * @Author liuxianzhao
 * @Date 2018年01月12日 14:45
 */
public class TestZKQueryData extends TestZK {
    @Test
    public void testGetChildrenSync() throws KeeperException, InterruptedException {
        ZooKeeper zk = ZooKeeperCreator.getZk("testGetChildrenSync");
        String path = "/testGetChildrenSync";
        zk.create(path,
                "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        zk.create(path + "/c1",
                "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL);
//        List<String> children = zk.getChildren(path, true); //默认watcher能监听到节点变化
//        List<String> children = zk.getChildren(path, false);//无法监听节点变化
//        List<String> children = zk.getChildren(path, new Watcher() {
//            @Override
//            public void process(WatchedEvent event) {
//                //不使用默认，手动写个watcher来监听数据变化
//                logger.info(event.toString());
//            }
//        });
        Stat oldStat = new Stat();
        List<String> children = zk.getChildren(path, true, oldStat);
        children.forEach(s -> logger.info(s));
        logger.info(oldStat.hashCode() + "|" + oldStat.toString());
//        zk.create(path + "/c2", "".getBytes(),
//                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        List<String> children_ = zk.getChildren(path, true, oldStat);
        children_.forEach(s -> logger.info(s));
        logger.info(oldStat.hashCode() + "|" + oldStat.toString());
        Thread.sleep(Integer.MAX_VALUE);
    }

    @Test
    public void testGetChildrenAsync() throws KeeperException, InterruptedException {
        ZooKeeper zk = ZooKeeperCreator.getZk("testGetChildrenAsync");
        String path = "/testGetChildrenAsync";
        zk.create(path,
                "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        zk.create(path + "/c1",
                "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL);
        final List<String> children = new ArrayList<>();
        final CountDownLatch downLatch = new CountDownLatch(1);
        zk.getChildren(path, true, new AsyncCallback.ChildrenCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, List<String> children_) {
                if (KeeperException.Code.OK.intValue() == rc) {
                    logger.info("获取成功: path-{},ctx-{}", path, ctx);
                    children.addAll(children_);
                } else {
                    logger.error("获取失败:" + KeeperException.create(KeeperException.Code.get(rc), path).getMessage());
                }
                downLatch.countDown();
            }
        }, "TestCTX"); //默认watcher能监听到节点变化
        downLatch.await();
        children.forEach(s -> logger.info(s));

        //这个方法会返回Stat
        final CountDownLatch downLatch_ = new CountDownLatch(1);
        zk.getChildren(path, false, new AsyncCallback.Children2Callback() {
            @Override
            public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat_) {
                if (KeeperException.Code.OK.intValue() == rc) {
                    logger.info("获取成功: path-{},ctx-{}", path, ctx);
                    children.forEach(s -> logger.info(s));
                } else {
                    logger.error("获取失败:" + KeeperException.create(KeeperException.Code.get(rc), path).getMessage());
                }
                downLatch_.countDown();
            }
        }, "TestCTX");
        Thread.sleep(Integer.MAX_VALUE);
    }

    @Test
    public void testGetDataSync() throws KeeperException, InterruptedException {
        ZooKeeper zk = ZooKeeperCreator.getZk("testGetDataSync");
        String path = "/testGetDataSync";
        zk.create(path, "中文".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        Stat stat = new Stat();
        byte[] data = zk.getData(path, false, stat);
        logger.info(new String(data));
        logger.info("{},{},{}", stat.getCzxid(), stat.getMzxid(), stat.getVersion());
        zk.setData(path, "123".getBytes(), -1);
        byte[] data_ = zk.getData(path, false, stat);
        logger.info(new String(data_));
        logger.info("{},{},{}", stat.getCzxid(), stat.getMzxid(), stat.getVersion());
        Thread.sleep(Integer.MAX_VALUE);
    }

    @Test
    public void testGetDataASync() throws KeeperException, InterruptedException {
        ZooKeeper zk = ZooKeeperCreator.getZk("testGetDataASync");
        String path = "/testGetDataASync";
        zk.create(path, "中文".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        Stat stat = new Stat();
        zk.getData(path, true, new AsyncCallback.DataCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
                logger.info(new String(data));
                logger.info("{},{},{}", stat.getCzxid(), stat.getMzxid(), stat.getVersion());
            }
        }, "TestCTX");
        zk.setData(path, "123".getBytes(), -1);
        byte[] data_ = zk.getData(path, false, stat);
        logger.info(new String(data_));
        logger.info("{},{},{}", stat.getCzxid(), stat.getMzxid(), stat.getVersion());
        Thread.sleep(Integer.MAX_VALUE);
    }
}
